<!DOCTYPE html>
<html lang="zh-cn">
<head>
   
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9301743.html">Spark（十二）SparkSQL简单使用</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2>一、SparkSQL的进化之路</h2>
<p>1.0以前：&nbsp; &nbsp;Shark</p>
<p>1.1.x开始：SparkSQL(只是测试性的)&nbsp; SQL</p>
<p>1.3.x:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;SparkSQL(正式版本)+Dataframe</p>
<p>1.5.x:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; SparkSQL 钨丝计划</p>
<p>1.6.x：&nbsp; &nbsp; &nbsp; &nbsp;SparkSQL+DataFrame+DataSet(测试版本)</p>
<p>&nbsp;2.x:</p>
<ul>
<li>&nbsp;&nbsp;&nbsp;&nbsp; SparkSQL+DataFrame+DataSet(正式版本)</li>
<li>&nbsp;&nbsp;&nbsp;&nbsp; SparkSQL:还有其他的优化</li>
<li>&nbsp;&nbsp;&nbsp;&nbsp; StructuredStreaming(DataSet)</li>
</ul>
<p><strong>Spark on Hive和Hive on Spark</strong></p>
<ul>
<li class="15">Spark on Hive：&nbsp;Hive只作为储存角色，<strong>Spark负责sql解析优化，执行。</strong></li>
<li class="15">Hive on Spark：<strong>Hive</strong>即作为存储又<strong>负责sql的解析优化，Spark负责执行。</strong></li>
</ul>
<h2>二、认识SparkSQL</h2>
<h3>2.1　什么是SparkSQL?</h3>
<p>spark SQL是spark的一个模块，主要用于进行结构化数据的处理。它提供的最核心的编程抽象就是DataFrame。</p>
<h3>2.2　SparkSQL的作用</h3>
<p>提供一个编程抽象（DataFrame） 并且作为分布式 SQL&nbsp;查询引擎</p>
<p>DataFrame：它可以根据很多源进行构建，包括：<strong>结构化的数据文件，hive中的表，外部的关系型数据库，以及RDD</strong></p>
<h3>2.3　运行原理</h3>
<p>将&nbsp;Spark SQL&nbsp;转化为&nbsp;RDD，&nbsp;然后提交到集群执行</p>
<h3>2.4　特点</h3>
<p>（1）容易整合</p>
<p>（2）统一的数据访问方式</p>
<p>（3）兼容 Hive</p>
<p>（4）标准的数据连接</p>
<h3>2.5　SparkSession</h3>
<p>SparkSession是Spark 2.0引如的新概念。SparkSession为用户提供了统一的切入点，来让用户学习spark的各项功能。&nbsp;<br />&emsp;&emsp;在spark的早期版本中，SparkContext是spark的主要切入点，由于RDD是主要的API，我们通过sparkcontext来创建和操作RDD。对于每个其他的API，我们需要使用不同的context。例如，对于Streming，我们需要使用StreamingContext；对于sql，使用sqlContext；对于Hive，使用hiveContext。但是随着DataSet和DataFrame的API逐渐成为标准的API，就需要为他们建立接入点。所以在spark2.0中，引入SparkSession作为DataSet和DataFrame API的切入点，SparkSession封装了SparkConf、SparkContext和SQLContext。为了向后兼容，SQLContext和HiveContext也被保存下来。&nbsp;<br />　　&nbsp;<br />　　SparkSession实质上是SQLContext和HiveContext的组合（未来可能还会加上StreamingContext），所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。</p>
<p>特点：</p>
<p><strong>　　 ----&nbsp;</strong><strong>为用户提供一个统一的切入点使用Spark&nbsp;各项功能</strong></p>
<p><strong>&nbsp; &nbsp; &nbsp;&nbsp;&nbsp;&nbsp;----&nbsp;</strong><strong>允许用户通过它调用&nbsp;DataFrame&nbsp;和&nbsp;Dataset&nbsp;相关 API&nbsp;来编写程序</strong></p>
<p><strong>&nbsp; &nbsp; &nbsp;&nbsp;&nbsp;&nbsp;----&nbsp;</strong><strong>减少了用户需要了解的一些概念，可以很容易的与&nbsp;Spark&nbsp;进行交互</strong></p>
<p><strong>&nbsp; &nbsp; &nbsp;&nbsp;&nbsp;&nbsp;----&nbsp;</strong><strong>与&nbsp;Spark&nbsp;交互之时不需要显示的创建&nbsp;SparkConf, SparkContext&nbsp;以及 SQlContext，这些对象已经封闭在&nbsp;SparkSession&nbsp;中</strong></p>
<h3><strong>2.7　<strong>DataFrames&nbsp; &nbsp;</strong></strong></h3>
<p>在Spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503195056657-1280315007.png" alt="" width="700" /></p>
<h2>三、RDD转换成为DataFrame</h2>
<p>使用spark1.x版本的方式</p>
<p>测试数据目录：spark/examples/src/main/resources（spark的安装目录里面）</p>
<p>people.txt</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503195656260-525846269.png" alt="" width="700" /></p>
<h3>3.1&nbsp;<strong>通过&nbsp;case class&nbsp;创建&nbsp;DataFrames（反射）</strong></h3>
<div class="cnblogs_code">
<pre>//定义case class，相当于表结构
case class People(var name:String,var age:Int)
object TestDataFrame1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDDToDataFrame").setMaster("local")
    val sc = new SparkContext(conf)
    val context = new SQLContext(sc)
    // 将本地的数据读入 RDD， 并将 RDD 与 case class 关联
    val peopleRDD = sc.textFile("E:\\666\\people.txt")
      .map(line =&gt; People(line.split(",")(0), line.split(",")(1).trim.toInt))
    import context.implicits._
    // 将RDD 转换成 DataFrames
    val df = peopleRDD.toDF
    //将DataFrames创建成一个临时的视图
    df.createOrReplaceTempView("people")
    //使用SQL语句进行查询
    context.sql("select * from people").show()
  }
}</pre>
</div>
<p>运行结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503202629907-1000361533.png" alt="" width="700" /></p>
<h3>3.2&nbsp;<strong>通过&nbsp;<span lang="EN-US">structType&nbsp;创建&nbsp;DataFrames（编程接口）</span></strong></h3>
<div class="cnblogs_code">
<pre>object TestDataFrame2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val fileRDD = sc.textFile("E:\\666\\people.txt")
    // 将 RDD 数据映射成 Row，需要 import org.apache.spark.sql.Row
    val rowRDD: RDD[Row] = fileRDD.map(line =&gt; {
      val fields = line.split(",")
      Row(fields(0), fields(1).trim.toInt)
    })
    // 创建 StructType 来定义结构
    val structType: StructType = StructType(
      //字段名，字段类型，是否可以为空
      StructField("name", StringType, true) ::
      StructField("age", IntegerType, true) :: Nil
    )
    /**
      * rows: java.util.List[Row],
      * schema: StructType
      * */
    val df: DataFrame = sqlContext.createDataFrame(rowRDD,structType)
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()
  }
}</pre>
</div>
<p>运行结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503202905865-1517500300.png" alt="" width="700" /></p>
<h3>3.3&nbsp;<strong>通过 json 文件<span lang="EN-US">创建&nbsp;DataFrames</span></strong></h3>
<div class="cnblogs_code">
<pre>object TestDataFrame3 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val df: DataFrame = sqlContext.read.json("E:\\666\\people.json")
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()
  }
}</pre>
</div>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503203759442-1628348320.png" alt="" width="700" /></p>
<h2>四、DataFrame的read和save和savemode</h2>
<h3>4.1　数据的读取</h3>
<div class="cnblogs_code">
<pre>object TestRead {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    //方式一
    val df1 = sqlContext.read.json("E:\\666\\people.json")
    val df2 = sqlContext.read.parquet("E:\\666\\users.parquet")
    //方式二
    val df3 = sqlContext.read.format("json").load("E:\\666\\people.json")
    val df4 = sqlContext.read.format("parquet").load("E:\\666\\users.parquet")
    //方式三，默认是parquet格式
    val df5 = sqlContext.load("E:\\666\\users.parquet")
  }
}</pre>
</div>
<h3>4.2　数据的保存</h3>
<div class="cnblogs_code">
<pre>object TestSave {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val df1 = sqlContext.read.json("E:\\666\\people.json")
    //方式一
    df1.write.json("E:\\111")
    df1.write.parquet("E:\\222")
    //方式二
    df1.write.format("json").save("E:\\333")
    df1.write.format("parquet").save("E:\\444")
    //方式三
    df1.write.save("E:\\555")

  }
}</pre>
</div>
<h3>4.3　数据的保存模式</h3>
<p>使用mode</p>
<div class="cnblogs_code">
<pre>df1.write.format("parquet").mode(SaveMode.Ignore).save("E:\\444")</pre>
</div>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503211638036-705055493.png" alt="" width="700" /></p>
<h2>五、数据源</h2>
<h3>5.1　数据源只json</h3>
<p>参考4.1</p>
<h3>5.2　数据源之parquet</h3>
<p>参考4.1</p>
<h3>5.3　数据源之Mysql</h3>
<div class="cnblogs_code">
<pre>object TestMysql {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TestMysql").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    val url = "jdbc:mysql://192.168.123.102:3306/hivedb"
    val table = "dbs"
    val properties = new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","root")
    //需要传入Mysql的URL、表明、properties（连接数据库的用户名密码）
    val df = sqlContext.read.jdbc(url,table,properties)
    df.createOrReplaceTempView("dbs")
    sqlContext.sql("select * from dbs").show()

  }
}</pre>
</div>
<p>运行结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503215248558-1665335084.png" alt="" width="700" /></p>
<h3>5.4　数据源之Hive</h3>
<h4>（1）准备工作</h4>
<p>在pom.xml文件中添加依赖</p>
<div class="cnblogs_code">
<pre>&lt;!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive --&gt;
        &lt;dependency&gt;
            &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
            &lt;artifactId&gt;spark-hive_2.11&lt;/artifactId&gt;
            &lt;version&gt;2.3.0&lt;/version&gt;
        &lt;/dependency&gt;</pre>
</div>
<p>开发环境则把resource文件夹下添加hive-site.xml文件，集群环境把hive的配置文件要发到$SPARK_HOME/conf目录下</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180504184547333-1552631388.png" alt="" width="700" /></p>
<div class="cnblogs_code">
<pre>&lt;configuration&gt;
        &lt;property&gt;
                &lt;name&gt;javax.jdo.option.ConnectionURL&lt;/name&gt;
                &lt;value&gt;jdbc:mysql://localhost:3306/hivedb?createDatabaseIfNotExist=true&lt;/value&gt;
                &lt;description&gt;JDBC connect string for a JDBC metastore&lt;/description&gt;
                &lt;!-- 如果 mysql 和 hive 在同一个服务器节点，那么请更改 hadoop02 为 localhost --&gt;
        &lt;/property&gt;
        &lt;property&gt;
                &lt;name&gt;javax.jdo.option.ConnectionDriverName&lt;/name&gt;
                &lt;value&gt;com.mysql.jdbc.Driver&lt;/value&gt;
                &lt;description&gt;Driver class name for a JDBC metastore&lt;/description&gt;
        &lt;/property&gt;
        &lt;property&gt;
                &lt;name&gt;javax.jdo.option.ConnectionUserName&lt;/name&gt;
                &lt;value&gt;root&lt;/value&gt;
                &lt;description&gt;username to use against metastore database&lt;/description&gt;
        &lt;/property&gt;
        &lt;property&gt;
                &lt;name&gt;javax.jdo.option.ConnectionPassword&lt;/name&gt;
                &lt;value&gt;root&lt;/value&gt;
        &lt;description&gt;password to use against metastore database&lt;/description&gt;
        &lt;/property&gt;
    &lt;property&gt;
                &lt;name&gt;hive.metastore.warehouse.dir&lt;/name&gt;
                &lt;value&gt;/hive/warehouse&lt;/value&gt;
                &lt;description&gt;hive default warehouse, if nessecory, change it&lt;/description&gt;
        &lt;/property&gt;  
&lt;/configuration&gt;</pre>
</div>
<h4>（2）测试代码</h4>
<div class="cnblogs_code">
<pre>object TestHive {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    sqlContext.sql("select * from myhive.student").show()
  }
}</pre>
</div>
<p>运行结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180504192745282-1160176093.png" alt="" width="700" /></p>
<h2>六、SparkSQL 的元数据</h2>
<h3 id="blogTitle0">1.1元数据的状态</h3>
<p>SparkSQL 的元数据的状态有两种：</p>
<p>1、in_memory,用完了元数据也就丢了</p>
<p>2、hive , 通过hive去保存的，也就是说，hive的元数据存在哪儿，它的元数据也就存在哪儿。</p>
<p>换句话说，<strong>SparkSQL的数据仓库在建立在Hive之上实现的</strong>。我们要用SparkSQL去构建数据仓库的时候，必须依赖于Hive。</p>
<h3 id="blogTitle1">2.2Spark-SQL脚本</h3>
<p>如果用户直接运行bin/spark-sql命令。会导致我们的元数据有两种状态：</p>
<p>1、in-memory状态：如果SPARK-HOME/conf目录下没有放置hive-site.xml文件，元数据的状态就是in-memory</p>
<p>2、hive状态：如果我们在SPARK-HOME/conf目录下放置了，hive-site.xml文件，那么默认情况下，spark-sql的元数据的状态就是hive.</p></div>

</body>
</html>
